﻿/*
-----------------------------------------------------------------------------
This source file is part of Cell Cloud.

Copyright (c) 2009-2012 Cell Cloud Team - cellcloudproject@gmail.com

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
-----------------------------------------------------------------------------
*/

#include "CellStableHeaders.h"
#include "CellTalkServiceDaemon.h"
#include "CellTalkService.h"
#include "CellTalkPacketDefinition.h"
#include "CellSession.h"
#include "CellMessage.h"
#include "CellPacket.h"
#include "CellLoggerManager.h"
#include "CellUtil.h"
#include "CellCryptologyImpl.h"
#include "CellInet4Address.h"
#include "CellInet6Address.h"
#include "CellSpeaker.h"
#include "CellBlockingAcceptor.h"

namespace cell
{

#define DAEMON_HEARTBEAT_PERIOD 1000
#define SESSION_TIMEOUT 90000

//-----------------------------------------------------------------------
TalkServiceDaemon::TalkServiceDaemon(TalkService* service)
	: Thread()
	, mService(service)
	, mRunning(false)
	, mSpinning(false)
	, mCount(0)
{
}
//-----------------------------------------------------------------------
TalkServiceDaemon::~TalkServiceDaemon(void)
{
}
//-----------------------------------------------------------------------
void TalkServiceDaemon::run(void)
{
	mRunning = true;
	long delay = 1000;
	uint64 startTime = 0;
	uint64 endTime = 0;
	uint64 detlaTime = 0;
	mSpinning = true;

	while (mSpinning)
	{
		// 计算时间误差
		detlaTime = endTime - startTime;
		if (detlaTime <= uint64(DAEMON_HEARTBEAT_PERIOD))
		{
			delay = uint64(DAEMON_HEARTBEAT_PERIOD) - detlaTime;
		}
		else
		{
			delay = uint64(DAEMON_HEARTBEAT_PERIOD) - (detlaTime % uint64(DAEMON_HEARTBEAT_PERIOD));
		}

		if (delay < 2)
			delay = 2;
		else if (delay > 1000)
			delay = 1000;

		Thread::sleep(delay);

		startTime = Util::currentTimeMillis();

		// 记录 Tick
		mService->mTick = startTime;

		if (NULL != mService->mAcceptor)
		{
			SessionList sessionList;

			{
				CELL_LOCK_MUTEX(mService->mSessionMutex);

				if (false == mService->mUnidentifiedSessions.empty())
				{
					uint64 time = startTime;
					TalkService::SessionCertificates::iterator iter = mService->mUnidentifiedSessions.begin();
					while (iter != mService->mUnidentifiedSessions.end())
					{
						TalkService::Certificate* cert = iter->second;
						if (cert->checked == false)
						{
							cert->checked = true;
							// 向未校验的 Session 发出校验请求。
							deliverChecking(iter->first, cert->plaintext, cert->key);
						}
						else
						{
							if (time - cert->time > (uint64)30000l)
							{
								// 超过 30 秒，将连接加入关闭列表。
								sessionList.push_back(iter->first);
							}
						}
						// 下一个
						++iter;
					}
				}
			}

			TalkService::SessionCertificates::iterator iter;

			if (false == sessionList.empty())
			{
				for (int i = 0, size = sessionList.size(); i < size; ++i)
				{
					Session* session = sessionList[i];
					bool hit = false;

					{
						CELL_LOCK_MUTEX(mService->mSessionMutex);

						iter = mService->mUnidentifiedSessions.find(session);
						if (iter != mService->mUnidentifiedSessions.end())
						{
							TalkService::Certificate* cert = iter->second;
							CELL_DELETE cert;
							mService->mUnidentifiedSessions.erase(iter);

							hit = true;

							char log[128] = {0x0};
							sprintf(log, "[CC] Talk service session timeout: %s:%d",
								session->getAddress().getHostAddress().c_str(), session->getAddress().getPort());
							LoggerManager::getSingleton().log(log);
						}
					}

					if (hit)
					{
						// 关闭连接
						mService->mAcceptor->close(session);
					}
				}

				// 清空列表
				sessionList.clear();
			}

			++mCount;
			if (mCount >= 10)
			{
				mCount = 0;

				{
					CELL_LOCK_MUTEX(mService->mSessionMutex);

					// 校验是否有超时的 Session，并将超时 Session 移除。
					if (false == mService->mSessionContexts.empty())
					{
						TalkService::SessionContexts::iterator iter, end = mService->mSessionContexts.end();
						for (iter = mService->mSessionContexts.begin(); iter != end; ++iter)
						{
							TalkSessionContext* ctx = iter->second;
							if (startTime - ctx->tickTime >= (uint64)SESSION_TIMEOUT)
							{
								sessionList.push_back(iter->first);
							}
						}
					}
				}

				if (false == sessionList.empty())
				{
					SessionList::iterator siter, send = sessionList.end();
					for (siter = sessionList.begin(); siter != send; ++siter)
					{
						Session* session = (*siter);
						bool hit = false;

						{
							CELL_LOCK_MUTEX(mService->mSessionMutex);

							TalkService::SessionContexts::iterator iter = mService->mSessionContexts.find(session);
							if (iter != mService->mSessionContexts.end())
							{
								hit = true;

								char log[256] = {0x0};
								sprintf(log, "[CC] Session %s/%d timeout (%d s).", session->getAddress().getHostAddress().c_str(),
									session->getAddress().getPort(), int(SESSION_TIMEOUT * 0.001));
								LoggerManager::getSingleton().log(log);
							}
						}

						if (hit)
						{
							// 关闭
							mService->mAcceptor->close(session);
						}
					}

					sessionList.clear();
				}
			}

			// 驱动 FlashAddon
			if (mService->mFlashAddon)
			{
				mService->mFlashAddon->tick(startTime);
			}
		}

		{
			CELL_LOCK_MUTEX(mService->mSpeakerMutex);

			// 检查是否有需要重新连接的 Speaker

			if (!mService->mLostSpeakers.empty())
			{
				std::vector<Speaker*> candidates;

				TalkService::Speakers::iterator iter = mService->mLostSpeakers.begin(),
					end = mService->mLostSpeakers.end();
				while (iter != end)
				{
					Speaker* sper = iter->second;

					// 判断是否需要开始执行重连

					if (0 == sper->getTimestamp())
					{
						// 10 秒后重连
						sper->setTimestamp(startTime + 10000);
					}
					else if (startTime >= sper->getTimestamp())
					{
						// 达到重连时间
						candidates.push_back(sper);
					}

					++iter;
				}

				if (!candidates.empty())
				{
					for (int i = 0, size = candidates.size(); i < size; ++i)
					{
						Speaker* sper = candidates[i];
						// 重置时间戳
						sper->setTimestamp(0);
						const InetAddress* addr = sper->getAddress();
					#if CELL_DEBUG_MODE
						char log[128] = {0x0};
						sprintf(log, "[CC] Retry call %s @ %s", sper->getCelletIdentifier().c_str(), addr->getHostAddress().c_str());
						LoggerManager::getSingleton().log(log);
					#endif // CELL_DEBUG_MODE
						// 进行重连
						mService->call(Inet4Address(addr->getHostAddress(), addr->getPort()), sper->getCelletIdentifier());
					}
					candidates.clear();
				}
			}
		}

		endTime = Util::currentTimeMillis();
	}

	// 结束
	mRunning = false;
}
//-----------------------------------------------------------------------
void TalkServiceDaemon::stop(void)
{
	mSpinning = false;
}
//-----------------------------------------------------------------------
void TalkServiceDaemon::deliverChecking(Session* session,
	const String& text, const String& key)
{
	char ciphertext[128] = {0x0};
	int length = Cryptology::getSingleton().simpleEncrypt(ciphertext,
		text.c_str(), text.length(), key.c_str());

	// 包格式：密文|密钥

	// 组装数据
	// 使用简单加密
	char tag[] = TPT_INTERROGATE;
	Packet packet(tag, 1);
	packet.appendSubsegment(ciphertext, length);
	packet.appendSubsegment(key.c_str(), 8);

	// 生成消息
	char data[256] = {0x0};
	int len = Packet::pack(data, &packet);
	Message message(data, len);
	session->write(&message);
}

} // end namespace cell
